package com.archgeek.scala.code

import com.google.common.collect.Queues

import java.util.concurrent.TimeUnit

/**
 *
 * @author pizhihui
 * @date 2024-01-03 13:49
 */
object ThreadDaemonUtils {


  private val exportGroupTask: java.util.concurrent.BlockingQueue[CreateBatchRequest] =  new java.util.concurrent.ArrayBlockingQueue[CreateBatchRequest](10000)


  def main(args: Array[String]): Unit = {

    println("hello world!!!!")

    Utils.startDaemonThread("export-group-thread") {
      println("提交人群包任务线程.......初始化")
      val logPrefix = "扫描人群包导入线程池"
      while (true) {

        val groupList = new java.util.HashSet[CreateBatchRequest]()
        try {
          println(s"${logPrefix}|等待人群包任务....groupList=${groupList.size()},exportGroupTask=${exportGroupTask.size()}")
          val n = Queues.drain(exportGroupTask, groupList, 1, 1, TimeUnit.MINUTES)
          println(s"${logPrefix}|消费人群包任务....${n}...groupList=${groupList.size()},exportGroupTask=${exportGroupTask.size()}")
        } catch {
          case e: Exception =>
            println(s"${logPrefix}|消费人群包任务异常...", e)
        }
        // 线程池提交任务
        if (groupList.size() > 0) {
         println(s"开始执行任务.......${groupList.iterator().next()}")
        }

      }
    }

    Utils.startDaemonThread("add-thread"){
      val request = new CreateBatchRequest()
      request.id = Option(123)
      request.queue = Option("hadoop6")
      exportGroupTask.offer(request)
    }

    //
    System.in.read()

  }


}
